//
// Created by Alex on 17/3/13.
//

#ifndef LIBCR_BUFFER_H
#define LIBCR_BUFFER_H

#include <vector>
#include <cstdint>
#include <cstdio>
#include <string>
#include <map>
#include <sstream>
#include <sys/_types/_iovec_t.h>
#include "libcr.h"

enum {
  MIN_MSG_BUFFER_SIZE = 1024,
  DEFAULT_MAX_MSG_BUFFER_SIZE = 128 * 1024,
  MAX_EXTEND_BUFFER_SIZE = 512 * 1024
};


/**
 * socket buffer:
 *
 *   | head block | data block | tail block |
 *
 *   head block: data which have been read
 *   data block: data which have not beed read
 *   tail block: space which can be appended
 */
struct MsgBuffer {
  /**
   * 内存块指针
   */
  char *m_heapBuffer;

  /**
   * 内存块大小
   */
  uint32_t m_heapSize;

  /**
   * 数据块在 内存块的 offset
   */
  uint32_t m_dataOffset;

  /**
   * 数据块大小
   */
  uint32_t m_dataSize;

  /**
   * 是否对内存块增大限制
   */
  bool m_isLimit;

  /**
   * 当 m_isLimit = true时, 内存块不能超过的大小
   */
  uint32_t m_bufferMaxLimit;


  MsgBuffer() :
      m_heapBuffer(nullptr),
      m_heapSize(0),
      m_dataOffset(0),
      m_dataSize(0),
      m_isLimit(false),
      m_bufferMaxLimit(DEFAULT_MAX_MSG_BUFFER_SIZE) {}

  virtual ~MsgBuffer() {
    Release();
  }

  /**
   * 数据指针
   * @return
   */
  inline const char *Data() const {
    return m_heapBuffer + m_dataOffset;
  }

  /**
   * 当前内存指针
   * @return
   */
  inline const char *Space() const {
    return m_heapBuffer;
  }

  /**
   * 当前数据大小
   * @return
   */
  inline uint32_t Size() const {
    return m_dataSize;
  }

  /**
   * 当前内存块的大小
   * @return
   */
  inline uint32_t Capacity() const {
    return m_heapSize;
  }

  /**
   * 头部剩余空间
   * @return
   */
  inline uint32_t RemainHeadCapacity() const {
    return m_dataOffset;
  }

  /**
   * 尾部剩余空间
   * @return
   */
  inline uint32_t RemainTailCapacity() const {
    return m_heapSize - m_dataOffset - m_dataSize;
  }

  /**
   * 空间是否满
   * @return
   */
  inline bool IsFull() const {
    return 0 == (RemainHeadCapacity() + RemainTailCapacity());
  }

  /**
   * 扩展空间是否有限制
   */
  inline void SetNoLimit() {
    m_isLimit = false;
  }

  /**
   * 设置最大buffer限制
   * @param sizeLimit
   * @return
   */
  inline bool SetBufferMaxLimit(uint32_t sizeLimit) {
    if (sizeLimit < MIN_MSG_BUFFER_SIZE ||
        sizeLimit < m_heapSize) {
      return false;
    }

    m_bufferMaxLimit = sizeLimit;
    m_isLimit = true;
    return true;
  }

  /**
   * 扩展buffer的大小
   * @param size
   * @return
   */
  int Reserve(uint32_t size) {
    int ret = 0;
    if (!IsInit()) {
      ret = InitBuffer(size);
    } else {
      ret = Expand(size);
    }
    return ret;
  }

  /**
   * 添加数据,返回已经添加数据的大小
   * @param data
   * @param size
   * @return
   */
  uint32_t Append(const void *data, uint32_t size) {
    AdjustSpaceForTail(size);
    uint32_t appendSize = size > RemainTailCapacity() ?
                          RemainTailCapacity() : size;

    AppendInternal((char *) data, appendSize);

    return appendSize;
  }

  /**
   * 添加 char 数据
   * @param val
   * @param size
   * @return
   */
  uint32_t Append(char val, uint32_t size) {
    AdjustSpaceForTail(size);
    uint32_t appendSize = size > RemainTailCapacity() ?
                          RemainTailCapacity() : size;

    memset(GetWritePos(), val, appendSize);
    m_dataSize += appendSize;

    return appendSize;
  }

  /**
   * 从 fd 读满buffer
   * @param fd
   * @param recvRet
   * @return
   */
  int RecvToBuffer(int fd, ssize_t &recvRet) {
    if (!IsInit()) {
      InitBuffer(MIN_MSG_BUFFER_SIZE);
    }

    uint32_t remain = RemainTailCapacity();
    if (remain == 0) {
      Marshal();

      if (0 == RemainTailCapacity()) {
        return -1;
      }
    }

    if (m_isLimit && m_dataSize >= m_bufferMaxLimit) { //如果有限制: 一次读满数据

      recvRet = cr_read(fd, GetWritePos(), remain);
      if (recvRet < 0) {
        return 0;
      }

      m_dataSize += recvRet;

      if (recvRet == remain) {
        Marshal();
        Expand(m_heapSize << 2);
      }

    } else { //尽可能读完 socket
      static thread_local char extendBuffer[MAX_EXTEND_BUFFER_SIZE];

      struct iovec vec[2];
      vec[0].iov_base = GetWritePos();
      vec[0].iov_len = remain;
      vec[1].iov_base = extendBuffer;
      vec[1].iov_len = MAX_EXTEND_BUFFER_SIZE;

      recvRet = cr_read(fd, vec, 2);
      if (recvRet < 0) {
        return 0;
      }

      m_dataSize += recvRet;
      if (recvRet >= remain) {
        size_t expandSize = std::max(recvRet - remain, (m_heapSize << 2));
        Marshal();

        bool limit = m_isLimit;
        m_isLimit = false;
        Expand(expandSize);
        m_isLimit = limit;

        if (recvRet > remain) {
          Append(extendBuffer, recvRet - remain);
        }
      }
    }

    return 1;
  }

  /**
   * 跳过 size
   * @param size
   * @return
   */
  uint32_t Skip(uint32_t size) {
    uint32_t ret = 0;

    if (size > m_dataSize) {
      ret = m_dataSize;
      m_dataOffset = 0;
      m_dataSize = 0;
    } else {
      ret = size;
      m_dataOffset += size;
      m_dataSize -= size;
    }
    return ret;
  }

  /**
   * 计算append指定大小的数据需要移动的bytes
   * @param size
   * @return
   */
  uint32_t GetAppendMoveBytes(uint32_t size) {
    //当前append 大于结尾使用空间
    if (nullptr != m_heapBuffer && size > RemainTailCapacity()) {

      //需要移动全部data
      if (RemainTailCapacity() + RemainHeadCapacity() >= size) {
        return m_dataSize;
      } else {
        return m_dataSize;
      }
    }
    return 0;
  }

private:
  inline bool IsInit() const {
    return nullptr != m_heapBuffer;
  }

  /**
   * 初始化内存块
   * @param size
   * @return
   */
  int InitBuffer(uint32_t size) {
    uint32_t allocSize = std::min(size, MIN_MSG_BUFFER_SIZE);

    if (allocSize & 0xFFF) {
      allocSize &= 0xFFF;
      allocSize += 0xFFF;
    }

    if (m_isLimit && allocSize > m_bufferMaxLimit) {
      allocSize = m_bufferMaxLimit;
    }

    m_heapBuffer = new char[allocSize];
    m_heapSize = allocSize;

    return 1;
  }

  /**
   * 调整buffer空间
   * @param size
   */
  void AdjustSpaceForTail(uint32_t size) {
    if (!IsInit()) {
      InitBuffer(size);

    } else if (size > RemainTailCapacity()) {
      if (RemainHeadCapacity() + RemainTailCapacity() >= size) {
        //将数据内存对齐到头
        Marshal();

      } else {
        if (Expand(size + m_heapSize)) {

          if (RemainHeadCapacity() > 0) {
            Marshal();
          }
        }

      }
    }
  }

  /**
   * 扩展内存块大小
   * @param size
   * @param copyOffet
   * @return
   */
  int Expand(uint32_t size, uint32_t copyOffet = 0) {
    if (m_isLimit && m_heapSize >= m_bufferMaxLimit) {
      return 1;
    }

    uint32_t allocSize = size;
    if (allocSize & 0xFFF) {
      allocSize &= 0xFFF;
      allocSize += 0xFFF;
    }

    if (m_isLimit && allocSize > m_bufferMaxLimit) {
      allocSize = m_bufferMaxLimit;
    }

    char *newAddr = new char[allocSize];
    memcpy(newAddr + copyOffet, Data(), Size());

    delete[] m_heapBuffer;
    m_heapBuffer = newAddr;
    m_heapSize = allocSize;
    m_dataOffset = copyOffet;
    return 0;
  }

  /**
   * 添加数据
   * @param src
   * @param size
   */
  void AppendInternal(const char *src, uint32_t size) {
    memcpy(GetWritePos(), src, size);
    m_dataSize += size;
  }

  /**
   * 当前可写入的位置
   * @return
   */
  inline char *GetWritePos() const {
    return m_heapBuffer + m_dataOffset + m_dataSize;
  }

  /**
   * 数据块向前移动
   * @param offset
   */
  void Marshal(uint32_t offset = 0) {
    memmove(m_heapBuffer + offset, Data(), Size());
    m_dataOffset = offset;
  }

  /**
   * 释放空间
   */
  void Release() {
    if (m_heapBuffer != nullptr) {
      delete m_heapBuffer;
      m_heapBuffer = nullptr;
    }

    m_heapSize = 0;
    m_dataOffset = 0;
    m_dataSize = 0;
  }

};

/**
 * 定长 buffer 链表: MsgBuffer_1 -> MsgBuffer_2 -> MsgBuffer_3 ....
 */
struct BufferList {
  typedef std::list<MsgBuffer *>::iterator BufferIter;

  enum {
    DEFAULT_MAX_MOVE_BYTES = 16 * 1024
  };

  /**
   * buffer 链表
   */
  std::list<MsgBuffer *> m_bufferList;

  /**
   * 当前读buffer
   */
  BufferIter m_readIter;

  /**
   * 当前写buffer
   */
  BufferIter m_writeIter;

  /**
   * buffer在append时最大memmove的大小
   */
  uint32_t m_maxMoveBytes;

  /**
   * buffer的最大限制
   */
  uint32_t m_bufferMaxLimit;

  /**
   * buffer list的 大小
   */
  uint32_t m_listSize;

  BufferList() :
      m_maxMoveBytes(DEFAULT_MAX_MOVE_BYTES),
      m_bufferMaxLimit(DEFAULT_MAX_MSG_BUFFER_SIZE) {

    InitBufferList();
  }

  void InitBufferList() {
    MsgBuffer *buffer = new MsgBuffer();
    buffer->SetBufferMaxLimit(m_bufferMaxLimit);
    m_bufferList.push_back(buffer);
    m_listSize = 1;

    m_readIter = m_bufferList.begin();
    m_writeIter = m_bufferList.begin();
  }

  /**
   * 释放所有内存
   */
  void Clear() {
    for (MsgBuffer *buffer : m_bufferList) {
      delete buffer;
    }
    m_bufferList.clear();
    m_listSize = 0;
  }

  virtual ~BufferList() {
    Clear();
  }

  /**
   * 当前 buffer 的数据
   * @return
   */
  const char *Data() const {
    MsgBuffer *buffer = *m_readIter;
    return buffer->Data();
  }

  /**
   * 当前 buffer 的大小
   * @return
   */
  uint32_t Size() const {
    MsgBuffer *buffer = *m_readIter;
    return buffer->Size();
  }

  /**
   * 当前 buffer 的结尾可使用空间
   * @return
   */
  uint32_t RemainTailCapacity() {
    MsgBuffer *buffer = *m_readIter;
    return buffer->RemainTailCapacity();
  }

  /**
   * 当前 buffer 是否滿了
   * @return
   */
  bool IsFull() {
    MsgBuffer *buffer = *m_readIter;
    return buffer->IsFull();
  }

  bool haveMoreBuffer() {
    BufferIter next(m_readIter);
    next++;
    return next != m_bufferList.end();
  }

  /**
   * 设置buffer的最大内存限制
   * @param sizeLimit
   */
  void SetBufferMaxLimit(uint32_t sizeLimit) {
    m_bufferMaxLimit = sizeLimit;
    for (MsgBuffer *buffer : m_bufferList) {
      buffer->SetBufferMaxLimit(m_bufferMaxLimit);
    }
  }

  /**
   * 向 buffer 添加可写数据
   * @param data
   * @param size
   * @return
   */
  int Append(const char *data, uint32_t size) {
    uint32_t appendRet = 0;

    while (size > 0) {
      MsgBuffer *buffer = *m_writeIter;
      uint32_t remainCapacity = buffer->RemainTailCapacity();

      if (buffer->GetAppendMoveBytes(size) > m_maxMoveBytes &&
          remainCapacity > 0) {

        appendRet = buffer->Append(data, remainCapacity);
        data = data + appendRet;
        size = size - appendRet;
      } else {

        appendRet = buffer->Append(data, size);
        data = data + appendRet;
        size = size - appendRet;
      }

      if (size > 0) {
        MsgBuffer *newBuffer = new MsgBuffer;
        newBuffer->SetBufferMaxLimit(m_bufferMaxLimit);
        uint32_t lastBuffSize = buffer->Capacity();

        if (size < lastBuffSize) {
          newBuffer->Reserve(lastBuffSize);
        }

        appendRet = newBuffer->Append(data, size);
        data += appendRet;
        size -= appendRet;

        m_bufferList.push_back(newBuffer);
        m_listSize += 1;
        ++m_writeIter;
      }
    }

    return 0;
  }

  /**
   * 通过 socket 读取数据
   * @param fd
   * @param recvRet
   * @return
   */
  int RecvToBuffer(int fd, ssize_t &recvRet) {
    MsgBuffer *buffer = *m_writeIter;
    int ret = buffer->RecvToBuffer(fd, recvRet);
    //没有空间了
    if (ret == -1) {

      MsgBuffer *newBuffer = new MsgBuffer;
      newBuffer->SetBufferMaxLimit(m_bufferMaxLimit);
      uint32_t lastBufSize = buffer->Capacity();
      newBuffer->Reserve(lastBufSize);

      ret = newBuffer->RecvToBuffer(fd, recvRet);
      if (ret == -1) {
        return ret;
      }

      m_bufferList.push_back(newBuffer);
      m_listSize += 1;
      ++m_writeIter;
    }
    return ret;
  }

  int WriteToFd(int fd, ssize_t &sendRet) {
    struct iovec vec[2];

    struct iovec *writeVec = vec;
    if (m_listSize > 2) {
      writeVec = new iovec[m_listSize];
    }

    size_t cnt = 0;
    for (MsgBuffer *buffer : m_bufferList) {
      writeVec[cnt].iov_base = (void *) buffer->Data();
      writeVec[cnt].iov_len = buffer->Size();
      cnt += 1;
    }

    sendRet = cr_writev(fd, writeVec, cnt);
    if (sendRet < 0) {
      return -1;
    }

    if (writeVec != vec) {
      delete writeVec;
    }

    Clear();
    InitBufferList();
    return 1;
  }

  /**
   * 最大移动的 buffer大小
   * @param bytes
   */
  void SetMaxMoveBytes(uint32_t bytes) {
    m_maxMoveBytes = bytes;
  }

  /**
   * 跳过 size 个 bytes
   * @param size
   * @return
   */
  uint32_t Skip(uint32_t size) {
    MsgBuffer *buffer = *m_readIter;
    uint32_t ret = buffer->Skip(size);

    if (0 == buffer->Size() && m_readIter != m_writeIter) {
      BufferIter tmp = m_readIter;
      ++m_readIter;
      m_bufferList.erase(tmp);
      m_listSize -= 1;
      delete buffer;
    }

    return ret;
  }

};

#pragma pack(1)
struct PacketHeader {
  uint16_t m_magic;
  uint32_t m_dataLen;
};
#pragma pack ()

#define HEADER_SIZE sizeof(PacketHeader)

struct OutputBufferProcessor {
  PacketHeader m_packetHeader;
  const uint16_t m_magic;

  OutputBufferProcessor(uint16_t magic) : m_magic(magic) {}

  ssize_t WriteData(int fd, uint32_t seq, void *data, size_t size) {
    m_packetHeader.m_dataLen = size;
    m_packetHeader.m_magic = m_magic;

    struct iovec vec[2];
    vec[0].iov_base = &m_packetHeader;
    vec[0].iov_len = HEADER_SIZE;
    vec[1].iov_base = data;
    vec[1].iov_len = size;

    return cr_writev(fd, vec, 2);
  }
};

struct PacketProcessor {
private:
  uint16_t m_headerReadBytes;
  uint32_t m_dataReadBytes;
  const uint16_t m_magic;
  bool m_isSplit;
  char m_packetHeader[HEADER_SIZE];
  std::string m_dataBuffer;

public:
  PacketProcessor(uint16_t magic) : m_magic(magic) {
    m_headerReadBytes = 0;
    m_isSplit = false;
    m_dataReadBytes = 0;
    memset(m_packetHeader, 0, sizeof(m_packetHeader));
  }

  virtual void HandlerPacket(PacketHeader *header, const char *data, size_t len) = 0;

  int OnReadComplete(BufferList &buffer) {
    while (buffer.Size() > 0) {

      uint32_t headerRemainBytes = HEADER_SIZE - m_headerReadBytes;
      if (headerRemainBytes > 0) {
        uint32_t copySize = std::min(headerRemainBytes, buffer.Size());
        memcpy(m_packetHeader + m_headerReadBytes, buffer.Data(), copySize);
        buffer.Skip(copySize);
        m_headerReadBytes += copySize;
      }

      //读完头
      if (HEADER_SIZE == m_headerReadBytes) {
        PacketHeader *header = (PacketHeader *) m_packetHeader;
        uint32_t dataLen = header->m_dataLen;

        if (dataLen <= 0 && header->m_magic == m_magic) {
          HandlerPacket(header, nullptr, 0);
          continue;
        }

        if (buffer.Size() > 0) {
          uint32_t remainBytes = dataLen - m_dataReadBytes;
          if (!m_isSplit) {

            if (buffer.Size() >= dataLen) {

              if (header->m_magic == m_magic) {
                HandlerPacket(header, buffer.Data(), dataLen);
              }
              buffer.Skip(dataLen);
              Clear();
              continue;

            } else { // buffer.Size() < dataLen
              if (m_dataBuffer.capacity() < dataLen) {
                m_dataBuffer.reserve(dataLen);
              }

              uint32_t copySize = buffer.Size();
              memcpy(&m_dataBuffer[m_dataReadBytes], buffer.Data(), copySize);
              buffer.Skip(copySize);
              m_dataReadBytes += copySize;
              m_isSplit = true;
            }

          } else { // m_isSplit = true
            uint32_t copySize = std::min(remainBytes, buffer.Size());
            memcpy(&m_dataBuffer[copySize], buffer.Data(), copySize);
            buffer.Skip(copySize);
            m_dataReadBytes += copySize;

            if (dataLen == m_dataReadBytes) {
              if (header->m_magic == m_magic) {
                HandlerPacket(header, &m_dataBuffer[0], dataLen);
              }
              Clear();
              continue;
            }
          }

        }
      }
    }

    return 0;
  }

  void Clear() {
    memset(m_packetHeader, 0, sizeof(m_packetHeader));
    m_headerReadBytes = 0;
    m_isSplit = false;

    if (m_dataBuffer.capacity() > 16 * 1024) {
      m_dataBuffer.resize(16 * 1024);
    }
    m_dataReadBytes = 0;
  }
};

int main(int argc, char **argv) {
  BufferList *list = new BufferList;
  std::string data = "Hello world!";

  for (size_t i = 0; i < 1000000; ++i) {
    list->Append(data.c_str(), data.size());
  }

  PacketProcessor *processor;
  while (1) {
    int ret = list->RecvToBuffer(-1, ret);
    if (ret > 0) {
      processor->OnReadComplete(*list);
    }
  }
}

#endif //LIBCR_BUFFER_H
